package wei.xinya;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.io.BufferedWriter;
import java.io.File;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;

public class Consumer {
    public static void main(String[] args) {
        if (args.length < 2) {
            System.out.println("用法: Java -jar kafka-jar-with-dependencies.jar <host:port> <topics>");
            System.out.println("host:port               kafka服务地址");
            System.out.println("topics                  订阅频道，多个频道用‘,’分隔");
            System.exit(-1);
        }

        String server = args[0];
        String topics = args[1];
        System.out.println("订阅的频道：");
        List<String> topicList = Arrays.stream(topics.split(",")).peek(System.out::println).collect(Collectors.toList());

        Properties props = new Properties();
        props.put("bootstrap.servers", server);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        // 获取当前时间戳作为 group.id
        props.put("group.id", "myConsumer_" + Calendar.getInstance().getTimeInMillis());

        // 可选设置属性
        props.put("enable.auto.commit", "true");
        // 自动提交offset,每1s提交一次
        props.put("auto.commit.interval.ms", "1000");
        // 获取方式：earliest 或 latest
        props.put("auto.offset.reset", "earliest");
        // 获取当前时间戳作为 client.id
        props.put("client.id", String.valueOf(Calendar.getInstance().getTimeInMillis()));

        try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(topicList);
            Map<String, BufferedWriter> writerMap = getWriterMap(topicList);

            while (true) {
                // 从服务器开始拉取数据
                ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<byte[], byte[]> record : records) {
                    String topic = record.topic();
                    byte[] key = record.key();
                    if (key == null || key.length == 0) {
                        key = new byte[]{};
                    }

                    byte[] value = record.value();
                    if (value == null || value.length == 0) {
                        value = new byte[]{};
                    }

                    System.out.printf("time = %s,topic = %s ,partition = %d,offset = %d, key = %s, value = %s\n",
                            Calendar.getInstance().getTimeInMillis(), topic, record.partition(),
                            record.offset(), new String(key), new String(value));

                    BufferedWriter writer = writerMap.get(topic);
                    writer.write(new String(record.value()) + "\n");
                    writer.flush();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static Map<String, BufferedWriter> getWriterMap(List<String> topicList) throws IOException {
        Map<String, BufferedWriter> writerMap = new HashMap<>();
        byte[] uft8bom = {(byte) 0xef, (byte) 0xbb, (byte) 0xbf};

        for (String topic : topicList) {
            File file = new File(topic + ".txt");
            BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(Files.newOutputStream(file.toPath()), StandardCharsets.UTF_8), 1024);
            writer.write(new String(uft8bom));

            writerMap.put(topic, writer);
        }
        return writerMap;
    }

}
